use tokio::task;
use reqwest::Client;
use std::{os::windows::thread, sync::{atomic::{AtomicUsize, Ordering}, Arc}};
use flume::{Sender, Receiver, bounded};
use futures::stream::{FuturesUnordered, StreamExt};
use std::time::Instant;

use crate::{config, generater, sender};

// 启动生产者任务
pub async fn start_producers(
    tx: Sender<reqwest::Request>
) -> FuturesUnordered<task::JoinHandle<()>> {
    let mut producers = FuturesUnordered::new();

    for index in 0..config::CONFIG.requests_config.len(){
        for _i in 0..config::CONFIG.requests_config[index].threads {
            let tx = tx.clone();
            producers.push(task::spawn(async move {
                generater::generate_requests(tx, index).await;
            }));
        }
    }
    
    producers
}

// 启动消费者任务
pub async fn start_consumers(
    rx: Receiver<reqwest::Request>,
    total_counter: Arc<AtomicUsize>,
    success_counter: Arc<AtomicUsize>,
    client: Arc<Client>
) -> FuturesUnordered<task::JoinHandle<()>> {
    let mut consumers = FuturesUnordered::new();
    for i in 0..50 {
        let rx = rx.clone();
        let total_counter = Arc::clone(&total_counter);
        let success_counter = Arc::clone(&success_counter);
        let client = Arc::clone(&client);
        consumers.push(task::spawn(async move {
            sender::send_requests_for_duration(rx, total_counter, success_counter, client).await;
        }));
    }
    consumers
}